package dyyx;

import java.util.Random;
import java.util.concurrent.atomic.AtomicLong;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPubSub;

public class PubAndSubTest {
	
	private static final String server = "127.0.0.1";
	private static final int port = 6379;

	private static final Random rand = new Random();
	private static final AtomicLong seq = new AtomicLong();
	private static final String channel = "channel001";

	public static void main(String[] args) throws Exception {
		System.out.println("hello,PubAndSubTest");
		// System.out.println(jedis.info());

		Thread thread = new PubThread();
		thread.start();

		// redis.clients.jedis.exceptions.JedisDataException: ERR only
		// (P)SUBSCRIBE / (P)UNSUBSCRIBE / PING / QUIT allowed in this context

		SubThread subthread1 = new SubThread("sub1");
		SubThread subthread2 = new SubThread("sub2");

		subthread1.start();
		subthread2.start();
	
		Utils.doSleep(1000000);

	    // jedis.close();
		
	}

	private static final String getMsg() {
		return "msg-" + seq.getAndIncrement() + "," + Utils.getTimeString();
	}

	private static class Sub extends JedisPubSub {
		private String name;

		public Sub(String name) {
			this.name = name;
		}

		public void onMessage(String channel, String message) {
			System.out.println(name + ":" + channel + ":" + message);
		}
	}

	private static class PubThread extends Thread {
		final Jedis jedis = new Jedis(server,port);
		
		public void run() {
			
			while (true) {
				Utils.doSleep(rand.nextInt(500));
				try {
					String msg = getMsg();
					jedis.publish(channel, msg);
					System.out.println("publish," + msg);
				} catch (Throwable e) {
					System.out.println("publish error," + e + "," + Utils.getTimeString());
				}
			}
		}
	}

	private static class SubThread extends Thread {
		private final String name;
		private final Jedis client;

		public SubThread(String name) {
			this.name = name;
			client = new Jedis(server,port);

		}

		public void run() {
			System.out.println("SubThread run," + name + "," + Utils.getTimeString()+","+client);
            Utils.doSleep(1000);
            // 注意 subscribe 方法 会阻塞
			client.subscribe(new Sub(name), channel);
			client.close();
			System.out.println("SubThread run done ," + name + "," + Utils.getTimeString()+","+client);

		}
	}

}
